package io.jnet.common.api.impl;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import io.jnet.common.api.ChannelContext;
import io.jnet.common.api.ReadHandler;
import io.jnet.common.decodec.DecodeResult;
import io.jnet.common.decodec.FrameDecodec;
import io.jnet.common.exception.EndOfStreamException;
import io.jnet.common.exception.NotFitProtocolException;
import io.jnet.common.util.AioListener;

public abstract class AbstractReadHandler implements ReadHandler
{
    protected final FrameDecodec              frameDecodec;
    protected final ByteBuf<?>                ioBuf;
    protected final ChannelContext            channelContext;
    protected final AsynchronousSocketChannel socketChannel;
    protected final AioListener               channelLisntener;
    protected volatile boolean                readPending = false;
    
    public AbstractReadHandler(AsynchronousSocketChannel socketChannel, FrameDecodec frameDecodec, ByteBuf<?> ioBuf, AioListener channelListener, ChannelContext channelContext)
    {
        this.socketChannel = socketChannel;
        this.frameDecodec = frameDecodec;
        this.ioBuf = ioBuf;
        this.channelLisntener = channelListener;
        this.channelContext = channelContext;
    }
    
    @Override
    public void completed(Integer read, Void nothing)
    {
        if (read == -1)
        {
            catchThrowable(EndOfStreamException.instance, channelContext);
            return;
        }
        ioBuf.addWriteIndex(read);
        try
        {
            decodecAndProcess();
            socketChannel.read(getWriteBuffer(), null, this);
        }
        catch (Throwable e)
        {
            catchThrowable(e, channelContext);
        }
    }
    
    @Override
    public void failed(Throwable exc, Void nothing)
    {
        catchThrowable(exc, channelContext);
    }
    
    public void decodecAndProcess() throws Throwable
    {
        while (true)
        {
            DecodeResult decodeResult = frameDecodec.decodec(ioBuf);
            switch (decodeResult.getType())
            {
                case LESS_THAN_PROTOCOL:
                    return;
                case BUF_NOT_ENOUGH:
                    ioBuf.compact().ensureCapacity(decodeResult.getNeed());
                    return;
                case NOT_FIT_PROTOCOL:
                    catchThrowable(NotFitProtocolException.instance, channelContext);
                    return;
                case NORMAL:
                    ByteBuf<?> packet = decodeResult.getBuf();
                    process(packet);
            }
        }
    }
    
    /**
     * 将iobuf的内容进行压缩，返回一个处于可写状态的ByteBuffer
     * 
     * @return
     */
    protected ByteBuffer getWriteBuffer()
    {
        ByteBuffer ioBuffer = ioBuf.nioBuffer();
        ioBuffer.position(ioBuffer.limit()).limit(ioBuffer.capacity());
        return ioBuffer;
    }
    
    protected void catchThrowable(Throwable e, ChannelContext context)
    {
        channelLisntener.catchException(e, context);
    }
    
    protected abstract void process(ByteBuf<?> buf) throws Throwable;
    
    public void registerRead()
    {
        if (readPending)
        {
            throw new UnsupportedOperationException();
        }
        readPending = true;
        channelLisntener.readRegister(channelContext);
        try
        {
            socketChannel.read(getWriteBuffer(), null, this);
        }
        catch (Exception e)
        {
            channelLisntener.catchException(e, channelContext);
        }
    }
    
}
